package com.zr.disruptor.impl;

import com.zr.domain.bar.entity.MarketDataEvent;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;

@Component
@Order(1)
public class DisruptorSource extends RichParallelSourceFunction<MarketDataEvent> {

    @Autowired
    private  DisruptorQueue disruptorQueue;

    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<MarketDataEvent> ctx) {
        while (isRunning) {
            MarketDataEvent pop = this.disruptorQueue.pop();
            ctx.collect(pop);
        }
    }
    @Override
    public void cancel() {
        isRunning = false;
    }

    @PreDestroy
    public void destroy() throws Exception {
        disruptorQueue.destroy();
    }
}
